package com.yunhe.transfer.kafka;

import com.fasterxml.jackson.core.type.TypeReference;
import com.google.protobuf.ByteString;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.MqttConsts;
import com.yunhe.common.model.data.AlarmDpuFileDTO;
import com.yunhe.transfer.proto.EigProto;
import com.yunhe.common.util.CsvUtil;
import com.yunhe.common.util.JsonUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

/**
 * 云端下发DPU告警文件Kafka消息监听
 * @author duanxiaoyong
 */
@Component
public class AlarmDpuFileHandler {

    @Autowired
    private MqttClient mqttClient;

    private TypeReference<AlarmDpuFileDTO> typeReference = new TypeReference<AlarmDpuFileDTO>(){};

    /**
     * 监听下发告警点号文件到DPU消息
     */
    @KafkaListener(topics = KafkaConsts.Topic.AlarmDpuFileRefresh, groupId = KafkaConsts.Group.AlarmDpuFile)
    public void consumer(ConsumerRecord<String, String> record) throws Exception {
        AlarmDpuFileDTO alarmDpuFileDTO = JsonUtil.readValue(record.value(), typeReference);
        List<Map<String, Object>> data = alarmDpuFileDTO.getData();
        String content = createCsvContent(data);
        String fileName = "alarm.csv";
        EigProto.pbFile pbFile = EigProto.pbFile.newBuilder().setFileName(new String(fileName.getBytes(StandardCharsets.UTF_8))).
                setFileContent(ByteString.copyFrom(content.getBytes(StandardCharsets.UTF_8))).build();
        List<String> beeIds = alarmDpuFileDTO.getBeeIds();
        for (String beeId : beeIds) {
            // 重置测点文件
            mqttClient.publish(MqttConsts.TOPIC.DPU_PtFile_Client.value(beeId), pbFile.toByteArray(), 1, false);
        }
    }

    /**
     * 生成CSV文件内容
     */
    private String createCsvContent(List<Map<String, Object>> data) throws Exception {
        String titles = "点号,名称,别名,是否离散,是否计算点,计算公式,变换公式,逆变换公式,变化公式,判零公式,单位," +
                "上限值,下限值,最大变化,最小变化,是否实时点,是否SOE,默认值,本地地址";
        String keys = "pid,title,alias,sfls,sfjs,formula,bhgs,nbhgs,bhuags,plgs,dw,upper,lower,zdbh,zxbh,sfssd,sfsoe,mrz,bddz";
        for (Map<String, Object> map : data) {
            map.put("alias", "alarm" + map.get("pid"));
            map.put("upper", "1");
            map.put("lower", "1");
            map.put("sfls", "FALSE");
            map.put("sfjs", "TRUE");
            map.put("bhgs", "");
            map.put("nbhgs", "");
            map.put("bhuags", "");
            map.put("plgs", "");
            map.put("dw", "");
            map.put("zdbh", "");
            map.put("zxbh", "");
            map.put("sfssd", "TRUE");
            map.put("sfsoe", "TRUE");
            map.put("mrz", "0");
            map.put("bddz", "");
        }
        return CsvUtil.createCsv(data, titles, keys);
    }
}
